package com.mindai.esper.monitor.inputadapters.rocketmq;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.impl.util.MsgConvertUtil;
import com.mindai.common.annotation.RocketMqListener;
import com.mindai.common.component.rocketmq.consumer.AbsTraceMessageListener;
import com.mindai.esper.monitor.biz.domain.assembler.impl.EsperRequestAdapterAssembler;
import com.mindai.esper.monitor.biz.enums.RocketQueues;
import com.mindai.esper.monitor.inputadapters.model.EsperRequestAdapter;
import com.mindai.esper.monitor.inputadapters.service.ComplexEventProcessingService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * 民加科风信息技术有限公司
 *
 * @Description: 支付消息监听
 * @Author: Guo Hao
 * @Date: 2017/5/22 16:27
 */
@RocketMqListener(topic= RocketQueues.SERVICE_TRADE_TOPIC,tag = "TAG_007_PAY_PROXY"+"||"+"TAG_008_PAY_WITH_HOLDING")
@Slf4j
public class PayGateWayInputeAdapterMqListener extends AbsTraceMessageListener {

    @Autowired
    ComplexEventProcessingService complexEventProcessingService;
    @Override
    protected Action consumeMessage(Message message, ConsumeContext consumeContext) {
        log.info("--PayGateWayInputeAdapterMqListener---start---Topic:" + message.getTopic() + "//Tag:" + message.getTag());
        log.info(MsgConvertUtil.bytes2String(message.getBody(), "UTF-8") + "///Receive: " + message);

        try {
            EsperRequestAdapter esperRequestAdapter = new EsperRequestAdapter();
            esperRequestAdapter = new EsperRequestAdapterAssembler().assemble(message, esperRequestAdapter);
            complexEventProcessingService.dealEsperInputData(esperRequestAdapter);
        } catch (Exception e) {
            log.error("消息处理异常：", e);
        }
        log.info("--PayGateWayInputeAdapterMqListener---end");
        return Action.CommitMessage;
    }
}
